package com.kafka.testp;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.Optional;

@Component
public class KafkaReceiver {

	private static Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);

	@Autowired
	private AdminClient adminClient;

    @PostConstruct
    public void createTopic() {
        //删除 分区

        //查询主题
        NewTopic topic = new NewTopic("hello3", 2, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));

    }

	@KafkaListener(topics = {"hello3"})
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info("----------------- record =" + record);
            logger.info("------------------ message =" + message);
        }

    }


    /****************************************************** 下面测试多个分组 模拟多个应用*/

    /**
     * 分组模拟多个 应用消费全量数据（一个分组对应一个应用）
     * @param record
     */
    @KafkaListener(topics = {"TOPIC"},groupId = "G1")
    public void listenos(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info("G1---- record =" + record);
            logger.info("G1-----message =" + message);
        }

    }


    /**
     * 分组模拟多个 应用消费全量数据（一个分组对应一个应用）
     * @param record
     */
    @KafkaListener(topics = {"TOPIC"},groupId = "G2")
    public void listenos2(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info("G2---- record =" + record);
            logger.info("G2-----message =" + message);
        }

    }

}